//
// Created by win10 on 2021/1/4.
//


#ifndef BACKHEADSHARE_EVENTEMITTERASYNC_H
#define BACKHEADSHARE_EVENTEMITTERASYNC_H

#include <unistd.h>
#include <sys/epoll.h>
#include <queue>

#include "EventEmitter.h"


// implementation 2
class EventEmitterAsync : public EventEmitter {
private:

    std::unordered_map<std::string, int> _event_fds;    // 事件名 -> fd
    std::unordered_map<int, std::function<void()>> _handlers; // fd -> handler
    std::queue<std::function<void()>> _active_handlers; // 待执行的handler
    int _epoll_fd;

public:
    EventEmitterAsync() {
        if ((_epoll_fd = epoll_create(1024)) < 0)
            perror("epoll_create failed.");
    }

    ~EventEmitterAsync() override = default;

    void emit(const std::string &event) final {
        // not thread-safe
        int write_fd = _event_fds.at(event);
        uint8_t msg;
        write(write_fd, &msg, sizeof(msg));
    }

    void on(const std::string &event, const std::function<void()> &handler) final {
        int event_pipe[2]{};
        if (pipe(event_pipe) < 0) perror("pipe failed.");
        int read_end = event_pipe[0];
        int write_end = event_pipe[1];
        _event_fds[event] = write_end;
        _handlers[read_end] = handler;

        epoll_event listen{};
        listen.data.fd = read_end;
        listen.events = EPOLLIN;
        if (epoll_ctl(_epoll_fd, EPOLL_CTL_ADD, read_end, &listen) < 0)
            perror("epoll add failed.");
    }

    [[noreturn]] void exec() {
        uint8_t buf;
        auto *event_array = new epoll_event[1024];
        while (true) {
            int n_fd;
            if ((n_fd = epoll_wait(_epoll_fd, event_array, 1024, -1)) < 0)
                perror("epoll_wait failed.");
            for (int i = 0; i < n_fd; ++i) {
                int read_fd = event_array[i].data.fd;
                if (read(read_fd, &buf, sizeof(buf)) != 1)
                    perror("read 1 byte error.");
                std::cout << "active fd: " << read_fd << std::endl;
                _active_handlers.emplace(_handlers.at(read_fd));
            }
            while (!_active_handlers.empty()) {
                auto handler = std::move(_active_handlers.front());
                _active_handlers.pop();
                handler(); // execute
            }
        }
    }
};


#endif //BACKHEADSHARE_EVENTEMITTERASYNC_H
